package com.edu.realtime.app.dwd.db;

import com.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author zcx
 * @create 2022-10-20 19:46
 * 交易域：取消订单事实表
 */
public class DwdTradeOrderCancelDetail {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2.检查点相关设置
        /*//2.1 启动检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 job取消后，检查点是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置两个检查点之间的最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://nameservice1:8020/edu/ck");
        //2.7 设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME","root");*/

        //TODO 3.从订单预处理表中读取数据  创建Flink的动态表
        tableEnv.executeSql("create table dwd_trade_order_pre_process(\n" +
                "id string,\n" +
                "order_id string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "date_id string,\n" +
                "create_time string,\n" +
                "final_amount string,\n" +
                "od_ts string,\n" +
                "user_id string,\n" +
                "province_id string,\n" +
                "operate_date_id string,\n" +
                "update_time string,\n" +
                "order_status string,\n" +
                "`type` string,\n" +
                "`old` string,\n" +
                "oi_ts string,\n" +
                "row_op_ts timestamp_ltz(3))" + MyKafkaUtil.getKafkaDDL(
                "dwd_trade_order_pre_process", "dwd_trade_cancel_detail_group"));

        //TODO 4.过滤出取消订单行为
        Table filteredTable = tableEnv.sqlQuery("select\n" +
                "id,\n" +
                "order_id,\n" +
                "user_id,\n" +
                "course_id,\n" +
                "course_name,\n" +
                "province_id,\n" +
                "date_id,\n" +
                "create_time,\n" +
                "final_amount,\n" +
                "oi_ts ts,\n" +
                "row_op_ts\n" +
                "from dwd_trade_order_pre_process \n" +
                "where `type`='update'\n" +
                "and `old`['order_status'] is not null\n" +
                "and order_status = '1003'");
        tableEnv.createTemporaryView("filtered_table", filteredTable);

        //TODO 5.创建动态表 和要写入的kafka主题进行映射
        tableEnv.executeSql("create table dwd_trade_order_pre_process(\n" +
                "id string,\n" +
                "order_id string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "date_id string,\n" +
                "cancel_time string,\n" +
                "final_amount string,\n" +
                "od_ts string,\n" +
                "user_id string,\n" +
                "province_id string,\n" +
                "operate_date_id string,\n" +
                "update_time string,\n" +
                "order_status string,\n" +
                "`type` string,\n" +
                "`old` string,\n" +
                "oi_ts string,\n" +
                "row_op_ts timestamp_ltz(3),\n" +
                "primary key(id) not enforced)"+ MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_cancel_detail"));

        //TODO 6.写入
        tableEnv.executeSql(
                "insert into dwd_trade_cancel_detail select * from filtered_table");
    }
}
